package com.liy.framework.common.kafka;

import com.liy.framework.common.kafka.vo.OperationLog;
import com.liy.framework.common.utils.JSONUtils;
import java.io.IOException;
import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class ConsumerServer {
    private static long count = 0L;

    public ConsumerServer() {
    }

    @KafkaListener(
            topics = {"${topic.test}"},
            group = "group01"
    )
    public void testMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        ++count;
        System.out.println("---收到消息：");
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println(count + "=count, ---topic:test--------------- message =" + message);
        }

    }

    @KafkaListener(
            topics = {"logsTopic"}
    )
    public void processMessage(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();

            try {
                OperationLog log = (OperationLog)JSONUtils.json2Object(message.toString(), OperationLog.class);
                System.out.println("----------------- record =" + record);
                System.out.println("------------------ message =" + message);
                System.out.println(log.getFunctionName());
            } catch (IOException var5) {
                var5.printStackTrace();
            }
        }

    }
}
